package com.icbc.dccsh.writter;

import java.util.Properties;
import java.util.concurrent.LinkedBlockingQueue;

import org.apache.commons.configuration.ConfigurationException;
import org.apache.commons.configuration.XMLConfiguration;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.utils.Utils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.icbc.dccsh.logpumper.IConfigurable;
import com.icbc.dccsh.reader.RedisReader;

import redis.clients.jedis.JedisPool;

public class KafkaWritter extends Thread implements IConfigurable{
	public static final Logger LOG = LoggerFactory.getLogger(KafkaWritter.class);
	
	private LinkedBlockingQueue<String> queue;
	private String _queueName;
	private String _address;
	private int _waittime;
	private int _retryThreshold;

	public KafkaWritter(LinkedBlockingQueue<String> queue) {
		this.queue = queue;
	}

	public void initialize(){
		// Nothing to do.
	}
	
	public void destructor(){
		// Nothing to do.
	}
	
	public void parseConfig(){
		XMLConfiguration config;
		try {
			config = new XMLConfiguration("conf.xml");
			this._queueName = config.getString("kafka.queuename", "logpumper");
			this._address = config.getString("kafka.address", "localhost:9092");
			this._waittime = config.getInt("kafka.waittime", 500);
			this._retryThreshold = config.getInt("kafka.retry", 10);
		} catch (ConfigurationException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}
	public void run() {
		Properties props = new Properties();
//		props.put("bootstrap.servers", this._address);
//		// props.put("zookeeper.connect","localhost:2181");
//		// props.put("group.id","test-group");
//		props.put("acks", "0");
//		props.put("retries", 0);
//		props.put("batch.size", 16384);
//		props.put("linger.ms", 1);
//		props.put("buffer.memory", 33554432);
//		props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
//		props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

		props.put("bootstrap.servers", this._address);
//		props.put("zookeeper.connect", "localhost:2181");
//		props.put("group.id", "test-group");
		props.put("acks", "all");
		props.put("retries", 0);
		props.put("batch.size", 16384);
		props.put("linger.ms", 1);
		props.put("buffer.memory", 33554432);
		props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
		props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
		
		Producer<String, String> producer = new KafkaProducer<String, String>(props);
		long now = System.currentTimeMillis();
		LOG.info("Started writting to kafka...");
		int retry = 1;
		while (true) {
			String event = queue.poll();
			if (event == null) {
				LOG.info(String.format("encounter null... sleep %dms..., retry %d times...", this._waittime, retry));
				Utils.sleep(this._waittime);
				if (retry == this._retryThreshold){
					break;
				}
				retry++;
				continue;
			}
			if (event.equals("_produced_end")) {
				break;
			}
			producer.send(new ProducerRecord<String, String>(this._queueName, event));
		}
		producer.close();
		long elapsed = System.currentTimeMillis() - now;
		LOG.info(String.format("end... cost %s ms...", elapsed));
	}

}
